package org.jfsm.concurrent;

import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.jfsm.log.LogSource;

/**
 * Assigns submitted work to Threads. Ensures no two threads work on the same
 * context at any given time. Ensures all works for the same context are
 * executed in the order they are submitted. Ensures all works that are not
 * waiting for their contexts at any given time are picked up by a thread in the
 * order they are submitted.
 * 
 * @author ooktay
 */
public final class ContextExecutor {
    private final LogSource logger = new LogSource();

    private final int numWorkers;

    private final ArrayBlockingQueue<Work> queue;

    private final Worker[] worker;

    private final class Worker implements Runnable {
        private final int id;
        private final Thread thread;

        private Worker(int id, ThreadFactory threadFactory) {
            this.id = id;

            thread = threadFactory.newThread(this);
            thread.start();
        }

        public final void run() {
            try {
                // WORKER loop
                for (;;) {
                    //logger.log("Work poll worker:", id);
                    Work work = queue.take();

                    //logger.log("Work found worker:", id, " context:", work.context);

                    // WORKING
                    for (;;) {
                        try {
                            work.run();
                        } catch (Exception e) {
                            logger.log("Exception context:", work.context, " runner:", work.runner,
                                    " e:", e);
                            e.printStackTrace();
                            System.exit(1);
                        }

                        // BACKLOG check
                        if (!work.context.release()) {
                            work = work.context.dequeue();

                            logger.log("Work backlog worker:", id, " context:", work.context);
                            continue;
                        } else {
                            break;
                        }
                    }
                }

            } catch (Exception e) {
                StringBuffer st = new StringBuffer(e.toString());
                for (Object trace : e.getStackTrace()) {
                    st.append("\n");
                    st.append(trace);
                }
                logger.log("Exception in worker:", id, "\n", st);

                try {
                    Thread.sleep(100);

                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                e.printStackTrace();
                System.exit(1);
            }
        }

    }

    private ThreadFactory threadFactory;

    public ContextExecutor() {
        this(Executors.defaultThreadFactory(), Runtime.getRuntime().availableProcessors(), 2048);
    }

    public ContextExecutor(int qSize) {
        this(Executors.defaultThreadFactory(), Runtime.getRuntime().availableProcessors(), qSize);
    }

    public ContextExecutor(int numThreads, int qSize) {
        this(Executors.defaultThreadFactory(), numThreads, qSize);
    }

    public ContextExecutor(ThreadFactory threadFactory, int numThreads, int qSize) {
        this.numWorkers = numThreads;
        this.threadFactory = threadFactory;

        queue = new ArrayBlockingQueue<Work>(qSize);
        worker = new Worker[numWorkers];
        start();
    }

    private void start() {
        for (int i = 0; i < numWorkers; i++) {
            worker[i] = new Worker(i, threadFactory);
        }
    }

    public final boolean submit(ConcurrentContext context, ConcurrentRunner runner) {
        if (context == null || runner == null) {
            throw new NullPointerException();
        }

        Work work = new Work(context, runner);

        if (work.context.lockOrEnqueue(work)) {

            if (queue.offer(work)) {
                //logger.log("Submited context:", work.context);
                return true;
            } else {
                logger.log("Couldn't Submit context:", work.context);

                try {
                    Thread.sleep(100);

                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                System.out.println("Couldn't Submit context:" + work.context + " runner:" + runner);
                System.exit(1);

                return false;
            }
        } else {
            //logger.log("Work ctx queued context:", work.context);
            return true;
        }
    }
}
